/*
异步工作器
*/
#include "buffer.hpp"
#include <mutex>
#include <condition_variable>
#include <functional>
#include <memory>
#include <thread>
#include <atomic>
#include<unordered_map>

namespace viclog
{
    using Func_t = std::function<void(Buffer &)>;
    enum class AsyncType
    {
        ASYNC_SAFE,  // 安全状态，缓冲区满则阻塞
        ASYNC_UNSAFE // 无限扩容（常用于测试）

    };
    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Func_t &cb, AsyncType looper_type = AsyncType::ASYNC_SAFE) : _looper_type(looper_type),
                                                                                       _stop(false),
                                                                                       _thread(std::thread(&AsyncLooper::threadEntry, this)),
                                                                                       _callBack(cb)
        {
        }
        ~AsyncLooper() { stop(); }
        void stop()
        {
            _stop = true;
            _cond_con.notify_all(); // 唤醒所有工作线程
            _thread.join();         // 等待线程退出
        }
        void push(const char *data, size_t len)
        {
            // 1.无限扩容（用于测试）2.固定大小(满了阻塞)
            std::unique_lock<std::mutex> lock(_mutex);
            // 条件变量控制，缓冲区空间大于数据长度则可以添加
            if (_looper_type == AsyncType::ASYNC_SAFE)
                _cond_con.wait(lock, [&]()
                               { return _pro_buf.writeAbleSize() >= len; });
            _pro_buf.push(data, len);
            // 唤醒一个消费者
            _cond_con.notify_one();
        }

    private:
        void threadEntry() // 线程入口函数
        {
            // 调用回调函数对消费缓冲区的数据进行处理  处理完后初始化缓冲区，交换缓冲区
            while (1)
            {
                // 1.判断生产缓冲区有无数据，有：交换到消费缓冲区 无：阻塞
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    if (_stop && _pro_buf.empty())
                        break;

                    // 有数据或者退出前被唤醒，则返回真继续向下运行，否则重新陷入休眠
                    _cond_con.wait(lock, [&]()
                                   { return _stop || !_pro_buf.empty(); });
                    _con_buf.swap(_pro_buf);
                    // 2.唤醒生产者（生产者也可能因为缓冲区满而阻塞）
                    if (_looper_type == AsyncType::ASYNC_SAFE)
                        _cond_pro.notify_all();
                }
                // 3.在被唤醒后对消费缓冲区进行数据处理
                _callBack(_con_buf);
                // 4.初始化消费缓冲区 重置消费缓冲区
                _con_buf.reset();
            }
        }

    private:
        AsyncType _looper_type;
        std::atomic<bool> _stop; // 工作停止标志
        // 生产消费缓冲区
        Buffer _pro_buf;
        Buffer _con_buf;
        std::mutex _mutex;
        // 生产消费的条件变量
        std::condition_variable _cond_pro;
        std::condition_variable _cond_con;
        std::thread _thread; // 异步工作器对应的线程
        Func_t _callBack;    // 对缓冲区数据进行处理的回调，有异步工作器使用者传入
    };

    

}